package org.offsync.util

import scala.collection.mutable.PriorityQueue

sealed class CompletionException extends Exception

object CompletionException extends CompletionException

class ScheduledExecutor extends DisposableBase {
  private val taskQueue = new PriorityQueue[ScheduledTask]()(ScheduledTask.priorityOrdering)
  
  def execute(task: => Unit): Unit = execute(Time.now)(task)

  def execute(task: Option[Exception] => Unit): Unit = execute(Time.now)(task)
  
  def execute(time: Time) = new Object {
    def apply(task: => Unit): Unit = {
      val scheduledTask = time.schedule(task)
      lock.synchronized {
        taskQueue += scheduledTask
        lock.notifyAll()
      }
    }
    def apply(task: Option[Exception] => Unit): Unit = {
      val scheduledTask = time.schedule(task(None))
      lock.synchronized {
        taskQueue += scheduledTask
        lock.notifyAll()
      }
    }
  }
  
  def run(): Unit = {
    try {
      while (true) {
        val readyTask = lock.synchronized {
          while (taskQueue.size == 0 || !taskQueue.max.expired) {
            lock.wait()
          }
          taskQueue.dequeue()
        }
        
        try {
          readyTask()
        } catch {
          case e if !e.isInstanceOf[CompletionException] => println("Unhandled exception in ScheduledExecutor")
        }
      }
    } catch {
      case e: CompletionException => super.onDispose()
    }
  }

  protected override def onDispose(): Unit = {
    val scheduledTask = Time.minValue.schedule { condition =>
      throw CompletionException
    }
    
    lock.synchronized {
      taskQueue += scheduledTask
      lock.notifyAll()
    }
  }
}
